import logging
from typing import Optional

from moto.ec2 import models as ec2_models
from moto.utilities.id_generator import Tags

from localstack.services.ec2.exceptions import (
    InvalidSecurityGroupDuplicateCustomIdError,
    InvalidSubnetDuplicateCustomIdError,
    InvalidVpcDuplicateCustomIdError,
)
from localstack.utils.id_generator import (
    ExistingIds,
    ResourceIdentifier,
    localstack_id,
)
from localstack.utils.patch import patch

LOG = logging.getLogger(__name__)


@localstack_id
def generate_vpc_id(
    resource_identifier: ResourceIdentifier,
    existing_ids: ExistingIds = None,
    tags: Tags = None,
) -> str:
    # We return an empty string here to differentiate between when a custom ID was used, or when it was randomly generated by `moto`.
    return ""


@localstack_id
def generate_security_group_id(
    resource_identifier: ResourceIdentifier,
    existing_ids: ExistingIds = None,
    tags: Tags = None,
) -> str:
    # We return an empty string here to differentiate between when a custom ID was used, or when it was randomly generated by `moto`.
    return ""


@localstack_id
def generate_subnet_id(
    resource_identifier: ResourceIdentifier,
    existing_ids: ExistingIds = None,
    tags: Tags = None,
) -> str:
    # We return an empty string here to differentiate between when a custom ID was used, or when it was randomly generated by `moto`.
    return ""


class VpcIdentifier(ResourceIdentifier):
    service = "ec2"
    resource = "vpc"

    def __init__(self, account_id: str, region: str, cidr_block: str):
        super().__init__(account_id, region, name=cidr_block)

    def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
        return generate_vpc_id(
            resource_identifier=self,
            existing_ids=existing_ids,
            tags=tags,
        )


class SecurityGroupIdentifier(ResourceIdentifier):
    service = "ec2"
    resource = "securitygroup"

    def __init__(self, account_id: str, region: str, vpc_id: str, group_name: str):
        super().__init__(account_id, region, name=f"sg-{vpc_id}-{group_name}")

    def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
        return generate_security_group_id(
            resource_identifier=self, existing_ids=existing_ids, tags=tags
        )


class SubnetIdentifier(ResourceIdentifier):
    service = "ec2"
    resource = "subnet"

    def __init__(self, account_id: str, region: str, vpc_id: str, cidr_block: str):
        super().__init__(account_id, region, name=f"subnet-{vpc_id}-{cidr_block}")

    def generate(self, existing_ids: ExistingIds = None, tags: Tags = None) -> str:
        return generate_subnet_id(
            resource_identifier=self,
            existing_ids=existing_ids,
            tags=tags,
        )


def apply_patches():
    @patch(ec2_models.subnets.SubnetBackend.create_subnet)
    def ec2_create_subnet(
        fn: ec2_models.subnets.SubnetBackend.create_subnet,
        self: ec2_models.subnets.SubnetBackend,
        *args,
        tags: Optional[dict[str, str]] = None,
        **kwargs,
    ):
        # Patch this method so that we can create a subnet with a specific "custom"
        # ID.  The custom ID that we will use is contained within a special tag.
        vpc_id: str = args[0] if len(args) >= 1 else kwargs["vpc_id"]
        cidr_block: str = args[1] if len(args) >= 1 else kwargs["cidr_block"]
        resource_identifier = SubnetIdentifier(
            self.account_id, self.region_name, vpc_id, cidr_block
        )

        # tags has the format: {"subnet": {"Key": ..., "Value": ...}}, but we need
        # to pass this to the generate method as {"Key": ..., "Value": ...}.  Take
        # care not to alter the original tags dict otherwise moto will not be able
        # to understand it.
        subnet_tags = None
        if tags is not None:
            subnet_tags = tags.get("subnet", tags)
        custom_id = resource_identifier.generate(tags=subnet_tags)

        if custom_id:
            # Check if custom id is unique within a given VPC
            for az_subnets in self.subnets.values():
                for subnet in az_subnets.values():
                    if subnet.vpc_id == vpc_id and subnet.id == custom_id:
                        raise InvalidSubnetDuplicateCustomIdError(custom_id)

        # Generate subnet with moto library
        result: ec2_models.subnets.Subnet = fn(self, *args, tags=tags, **kwargs)
        availability_zone = result.availability_zone

        if custom_id:
            # Remove the subnet from the default dict and add it back with the custom id
            self.subnets[availability_zone].pop(result.id)
            old_id = result.id
            result.id = custom_id
            self.subnets[availability_zone][custom_id] = result

            # Tags are not stored in the Subnet object, but instead stored in a separate
            # dict in the EC2 backend, keyed by subnet id.  That therefore requires
            # updating as well.
            if old_id in self.tags:
                self.tags[custom_id] = self.tags.pop(old_id)

        # Return the subnet with the patched custom id
        return result

    @patch(ec2_models.security_groups.SecurityGroupBackend.create_security_group)
    def ec2_create_security_group(
        fn: ec2_models.security_groups.SecurityGroupBackend.create_security_group,
        self: ec2_models.security_groups.SecurityGroupBackend,
        name: str,
        *args,
        vpc_id: Optional[str] = None,
        tags: Optional[dict[str, str]] = None,
        force: bool = False,
        **kwargs,
    ):
        vpc_id = vpc_id or self.default_vpc.id
        resource_identifier = SecurityGroupIdentifier(
            self.account_id, self.region_name, vpc_id, name
        )
        custom_id = resource_identifier.generate(tags=tags)

        if not force and self.get_security_group_from_id(custom_id):
            raise InvalidSecurityGroupDuplicateCustomIdError(custom_id)

        # Generate security group with moto library
        result: ec2_models.security_groups.SecurityGroup = fn(
            self, name, *args, vpc_id=vpc_id, tags=tags, force=force, **kwargs
        )

        if custom_id:
            # Remove the security group from the default dict and add it back with the custom id
            self.groups[result.vpc_id].pop(result.group_id)
            old_id = result.group_id
            result.group_id = result.id = custom_id
            self.groups[result.vpc_id][custom_id] = result

            # Tags are not stored in the Security Group object, but instead are stored in a
            # separate dict in the EC2 backend, keyed by id.  That therefore requires
            # updating as well.
            if old_id in self.tags:
                self.tags[custom_id] = self.tags.pop(old_id)

        return result

    @patch(ec2_models.vpcs.VPCBackend.create_vpc)
    def ec2_create_vpc(
        fn: ec2_models.vpcs.VPCBackend.create_vpc,
        self: ec2_models.vpcs.VPCBackend,
        cidr_block: str,
        *args,
        tags: Optional[list[dict[str, str]]] = None,
        is_default: bool = False,
        **kwargs,
    ):
        resource_identifier = VpcIdentifier(self.account_id, self.region_name, cidr_block)
        custom_id = resource_identifier.generate(tags=tags)

        # Check if custom id is unique
        if custom_id and custom_id in self.vpcs:
            raise InvalidVpcDuplicateCustomIdError(custom_id)

        # Generate VPC with moto library
        result: ec2_models.vpcs.VPC = fn(
            self, cidr_block, *args, tags=tags, is_default=is_default, **kwargs
        )
        vpc_id = result.id

        if custom_id:
            # Remove security group associated with unique non-custom VPC ID
            default = self.get_security_group_from_name("default", vpc_id=vpc_id)
            if not default:
                self.delete_security_group(
                    name="default",
                    vpc_id=vpc_id,
                )

            # Delete route table if only main route table remains.
            for route_table in self.describe_route_tables(filters={"vpc-id": vpc_id}):
                self.delete_route_table(route_table.id)  # type: ignore[attr-defined]

            # Remove the VPC from the default dict and add it back with the custom id
            self.vpcs.pop(vpc_id)
            old_id = result.id
            result.id = custom_id
            self.vpcs[custom_id] = result

            # Tags are not stored in the VPC object, but instead stored in a separate
            # dict in the EC2 backend, keyed by VPC id.  That therefore requires
            # updating as well.
            if old_id in self.tags:
                self.tags[custom_id] = self.tags.pop(old_id)

            # Create default network ACL, route table, and security group for custom ID VPC
            self.create_route_table(
                vpc_id=custom_id,
                main=True,
            )
            self.create_network_acl(
                vpc_id=custom_id,
                default=True,
            )
            # Associate default security group with custom ID VPC
            if not default:
                self.create_security_group(
                    name="default",
                    description="default VPC security group",
                    vpc_id=custom_id,
                    is_default=is_default,
                )

        return result
